//package cn.ac.iie.di.ban.nmpp.handler;
//
//import cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv;
//import cn.ac.iie.di.ban.nmpp.tools.MppTools;
//import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
//import com.alibaba.fastjson.JSONObject;
//import com.datastax.driver.core.*;
//import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy;
//import com.datastax.driver.core.policies.DefaultRetryPolicy;
//import org.apache.commons.codec.DecoderException;
//import org.apache.commons.codec.binary.Hex;
//import org.codehaus.jackson.map.ObjectMapper;
//
//import java.io.IOException;
//import static java.lang.Thread.sleep;
//import java.nio.ByteBuffer;
//import java.util.*;
//
//import org.apache.log4j.Logger;
//import static cn.ac.iie.di.ban.nmpp.commons.RuntimeEnv.UPDATE_FORWARDMQ;
//import static com.datastax.driver.core.DataType.Name.SET;
//import com.datastax.driver.core.exceptions.InvalidQueryException;
//import com.datastax.driver.core.exceptions.NoHostAvailableException;
//import com.datastax.driver.core.exceptions.OperationTimedOutException;
//import com.datastax.driver.core.exceptions.QueryExecutionException;
//
///**
// * Created by jinsheng on 16/9/18.
// *
// */
//public class SE2DWWorkerMPPUpdateXW extends SE2DWWorker {
//
//    private String tableName;
//    private String utableName;
//    private String timestampKey = "";
//    private String timestampFieldKey;
//    private String keySpace;
//    long startTime = -1;
////rocketmq Type
//    private Map<String, REFieldType> rtn;
//    private List<String> updateFields;
//    static Map<String, String> ColumnNameType = new HashMap<String, String>();
//    private List<ColumnMetadata> columns;
//    private List<ColumnMetadata> columns2;
//    private Map<String, Type> columnMap = new HashMap<>();
//    private Map<String, Type> columnMap2 = new HashMap<>();
//
//    private final ObjectMapper objectMapper = new ObjectMapper();
//    private static org.apache.log4j.Logger logger = Logger.getLogger(SE2DWWorkerMPPUpdateXW.class.getName());
//
//    private Cluster cluster;
//    private Session session;
////    updateInsert
//    private Cluster clusterUI;
//    private Session sessionUI;
//
//    private String updatePrimaryKey;
//    private PreparedStatement insert_statement = null;
//    private PreparedStatement update_statment = null;
//
//    public SE2DWWorkerMPPUpdateXW(TableSe2DWHandler pTableSe2DWHandler, String pDWType) {
//        super(pTableSe2DWHandler, pDWType);
//    }
//
//    public List<ColumnMetadata> getColumns() {
//        return columns;
//    }
//
//    @Override
//    public void init() throws Exception {
//        tableName = tableSe2DBRule.getTableName();
//        if (UPDATE_FORWARDMQ.get(tableName) != null) {
//            updateFields = UPDATE_FORWARDMQ.get(tableName).getColumnSet();
//            updatePrimaryKey = UPDATE_FORWARDMQ.get(tableName).getQueryid();
//            logger.info(" " + updatePrimaryKey + " " + updateFields);
//        }
//        initPStatement();
//        uiPStatement();
//    }
//
//    private void initConnection() {
//
//        if (cluster != null) {
//            return;
//        }
//        int readTimeOut = 12000;
//        Boolean retry = true;
//        for (int tryTimes = 0;; tryTimes++) {
//            logger.info(tryTimes + "st times try get available connection to " + tableSe2DBRule.getDbType() + " dw");
//            String[] dbmeta = tableSe2DBRule.getConnStr().split("[/]");
//            if (dbmeta.length != 3) {
//                logger.error("unknow dbtype " + tableSe2DBRule.getDbType() + " dw unsuccessfully for " + tableSe2DBRule.getConnStr());
//            }
//            String userName = tableSe2DBRule.getUserName();
//            String password = tableSe2DBRule.getPassword();
//
//            QueryOptions options = new QueryOptions();
//            options.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
//            Cluster.Builder builder = Cluster.builder();
//            String dataCenter = dbmeta[2];
//            keySpace = dbmeta[1];
//            tableName = tableSe2DBRule.getTableName();
//            String coordinatorAddress = dbmeta[0];
//            //创建连接
//            //需要用户认证
//            if (userName.length() > 0) {
//                builder
//                        .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
//                        .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
//                        //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
//                        .withCredentials(userName, password).addContactPoint(coordinatorAddress).withQueryOptions(options);
//                // 不需要用户认证{
//            } else {
//                builder
//                        .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
//                        .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dataCenter).build())
//                        //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
//                        .addContactPoint(coordinatorAddress).withQueryOptions(options);
//            }
//            if (retry == null || retry) {
//                this.cluster = builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE).build();
//            } else {
//                this.cluster = builder.build();
//            }
//            break;
//
//        }
//
//    }
//
//    private void uiPStatement() {
//
//        if (clusterUI != null || sessionUI != null || insert_statement != null) {
//            return;
//        }
//
//        for (int tryTimes = 0;; tryTimes++) {
//            try {
//
//                String UI = UPDATE_FORWARDMQ.get(tableName).getForwardMq();
//                String dc = "";
//                String ks = "";
//                String tb = "";
//                String ip = "";
//                String username = "";
//                String passwd = "";
//                String[] updateTableConf = UI.split(":");
//                if (updateTableConf.length == 5) {
//                    ip = updateTableConf[0];
//                    ks = updateTableConf[1];
//                    dc = updateTableConf[2];
//                    utableName = updateTableConf[3];
//                    if (updateTableConf[4].split(";").length != 2) {
//                        logger.info("update sname is error" + updateTableConf[4]);
//                        System.exit(0);
//                    }
//                    username = updateTableConf[4].split(";")[0];
//                    passwd = updateTableConf[4].split(";")[1];
//                    tb = updateTableConf[3];
////            10.136.131.130:data_update:data-update:tp_msg_wechat_public_interactive_ulog:cassandra;cassandra
//                } else {
//                    logger.info("update table is null");
//                    System.exit(0);
//
//                }
//                int readTimeOut = 12000;
//                Boolean retry = true;
//                //�创建链接
//                QueryOptions options = new QueryOptions();
//                options.setConsistencyLevel(ConsistencyLevel.LOCAL_ONE);
//                Cluster.Builder builder = Cluster.builder();
//
//                builder
//                        .withSocketOptions(new SocketOptions().setConnectTimeoutMillis(120000).setReadTimeoutMillis(readTimeOut))
//                        .withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(dc).build())
//                        //.withRetryPolicy(DefaultRetryPolicy.INSTANCE)
//                        .withCredentials(username, passwd).addContactPoint(ip).withQueryOptions(options);
//
//                //设置重试策略�
//                if (retry == null || retry) {
//                    this.clusterUI = builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE).build();
//                } else {
//                    this.clusterUI = builder.build();
//                }
////绑定keyspace
//                this.sessionUI = this.clusterUI.connect(ks);
//                //获取该表元数据Ԫ���
//                this.columns2 = this.clusterUI.getMetadata().getKeyspace(ks).getTable(utableName).getColumns();
//                for (ColumnMetadata name : columns2) {
//                    columnMap2.put(name.getName(), resolveType(name.getType()));
//                }
//                logger.info(columnMap2.toString());
//                if (insert_statement == null) {
//                    // 预编译insert语句
//                    insert_statement = sessionUI.prepare("INSERT INTO " + tb + " JSON :json");
//                    logger.info("INSERT INTO " + tb + " JSON :json");
//                }
//                break;
//            } catch (Exception E) {
//                insert_statement = null;
//                session = null;
//                clusterUI = null;
//                E.printStackTrace();
//                logger.error(E);
//            }
//        }
//    }
//
//    public void initPStatement() {
//
//        if (cluster == null) {
//            for (int tryTimes = 0;; tryTimes++) {
//                logger.info(tryTimes + "st times try get valid pstmt to table " + tableSe2DBRule.getTableName() + " on " + tableSe2DBRule.getDbType() + " dw");
//                initConnection();
//                if (cluster == null) {
//                    logger.error("fatal error when getting connection to " + tableSe2DBRule.getDbType() + " dw");
//                } else {
//                    try {
//                        logger.info("create pstmt...");
//                        long startTime = System.nanoTime();
//                        //获取该表的元数据����
//                        this.session = this.cluster.connect(keySpace);
//                        //处理每一列的类型并保存，用于更新时确定调用哪个set方法
//                        logger.info(keySpace + " " + tableName);
//                        this.columns = this.cluster.getMetadata().getKeyspace(keySpace).getTable(tableSe2DBRule.getTableName()).getColumns();
//                        for (ColumnMetadata name : columns) {
//                            columnMap.put(name.getName(), resolveType(name.getType()));
//                        }
//                        logger.info(columnMap.toString());
//                        long endTime = System.nanoTime();
//                        logger.info("create pstmt successfully using " + (endTime - startTime) / 1000000L + " ms");
//                        logger.info("checkmongodb count and " + tableSe2DBRule.getTableName() + " count is   ");
//
//                        break;
//                    } catch (Exception ex) {
//                        logger.error("create pstmt to " + tableSe2DBRule.getTableName() + " unsuccessfully for " + ex.getMessage(), ex);
//                        try {
//                            if (session != null) {
//                                session.close();
//                                cluster = null;
//                            }
//                        } catch (Exception ex2) {
//                        } finally {
//                            update_statment = null;
//                            session = null;
//                            cluster = null;
//
//                        }
//                    }
//                }
//                try {
//                    Thread.sleep(2000);
//                } catch (Exception ex) {
//                }
//            }
//        }
//    }
//
//    @Override
//    public void loadData0() {
//        try {
//            if (docRecordSet.size() < 1) {
//                logger.warn("no record load to table " + tableSe2DBRule.getTableName() + " in " + dwType);
//                return;
//            }
//            int validRecordNum = 0;
//            int tryTimes = 0;
//            boolean failed = false;
//            long endTime = -1;
//            for (tryTimes = 0; tryTimes < 10; tryTimes++) {
//                logger.info(tryTimes + "st try to write " + docRecordSet.size() + " records to " + tableSe2DBRule.getTableName() + "...");
//                initPStatement();
//                logger.info(tryTimes + "st set parameters...");
//                String columnName = "";
//                String rocketMqColumnName = "";
//                ArrayList<Map<String, Object>> DataR = new ArrayList<Map<String, Object>>();
//                if (rtn == null) {
//                    rtn = RuntimeEnv.ROCKETMQ_DOC_DESC.get(tableName);
//                }
//                startTime = System.nanoTime();
//                
//                for (HashMap docRecord : docRecordSet) {
//                    Map<String, Object> newSingleData = new HashMap<String, Object>();
//                    for (String fields : updateFields) {
//                        columnName = fields;
//                        rocketMqColumnName = GetColumnName(columnName);
//                        Object value = docRecord.get(rocketMqColumnName);
//                        if (docRecord.get(rocketMqColumnName) != null) {   
//                            newSingleData.put(fields, value);
//                        }
//                    }
//                    DataR.add(newSingleData);
//                    validRecordNum++;
//                }
//                logger.info("data begin loader newmpp and size is " + DataR.size());
//                try {
//                    while (!updateData(DataR)) {
//                        logger.error("insert error  and sleep" + 500 + "ms");
//                        sleep(500);
//                        close();
//                        initPStatement();
//                    };
//
//                } catch (Exception E) {
//                    logger.error(E.getMessage(),E);
//                    continue;
//                }
//                docRecordSet.clear();
//                break;
//            }
//        } catch (Exception E) {
//            logger.error(E);
//
//        }
//    }
//
//    private boolean updateData(List<Map<String, Object>> lists) throws InterruptedException {
//        List<Map<String, Object>> succ = new ArrayList<Map<String, Object>>();
//        succ.addAll(lists);
//        if (lists.size() < 1) {
//            return true;
//        }
//        boolean retu = true;
//        long st = System.currentTimeMillis();
//        List<ResultSet> future = new ArrayList<>();
//        for (Map<String, Object> map : lists) {
//            if (map.size() <= 1) {
//                continue;
//            }
//            // 预编译update语句
//            Set<String> feilds = map.keySet();
//            String sql = getUpdatesql(updatePrimaryKey, feilds);
//            logger.info("update sql is  " + sql);
//            logger.info(map.toString());
//            update_statment = session.prepare(sql);
//            //绑定更新数据，并创建BoundStatement对象
//            BoundStatement bs = update_statment.bind();  
//            setUpdateValue(map, bs,feilds);
//            // 绑定更新记录的主�?
//            columnMap.get(updatePrimaryKey).bind("key", map.get(updatePrimaryKey), bs);
//            Integer seqn = 0;
//            columnMap.get("seqn").bind("seqn", seqn, bs);
////            setKeys(updatePrimaryKey, map, bs);
//
//            // 设置记录时间�?
//            if (this.timestampKey.length() > 0 && map.containsKey(this.timestampKey)) {
//                bs.setDefaultTimestamp((Long) map.get(this.timestampKey) * 1000000); // second to microsecond
//            }
//            try {
//                future.add(session.execute(bs));
//                logger.info(" hahah ");
//            } catch (Exception E) {
//                logger.info(E.getMessage(),E);
//                retu = false;
//            }
//        }
//        // 处理执行结果
//        int update = 0;
//        for (int i = 0; i < future.size(); i++) {
//            try {
//                ResultSet rs = future.get(i);
//                String result_string = rs.toString();
//                if (result_string.contains("ERROR")) {
//                    logger.error(result_string);
//                } else if (rs.wasApplied()) {
//                    //更新成功
////                    succ.add(lists.get(i));
//                    update++;
//                } else {
//                    logger.info(result_string);
//                    logger.info(lists.get(i));
////                    retu = false;
//                    // TODO: 2017-03-25 此处可能是由于记录不存在等原因�?成更新未能生�?by linchan
//                }
//
//            } catch (Exception e) {
//                logger.info(e.getMessage(),e);
//                retu = false;
//                // 异常捕捉�?30~234�?
//            }
//        }
//        future.clear();
//        logger.info("0st load " + lists.size() + "|expect: " + update + " records update to table " + tableSe2DBRule.getTableName() + " and mqName is " + tableSe2DBRule.getMqName() + " successfully using " + (System.currentTimeMillis() - st) / 1000000L + " ms " + retu);
//        while (true) {
//
//            try {
//                while (!push_single_json_data(succ)) {
//                    logger.error("insert error  and sleep" + 500 + "ms");
//                    sleep(500);
//                    uclose();
//                    uiPStatement();
//                };
//                succ.clear();
//            } catch (Exception E) {
//                logger.error("insertsss  " + E);
//                sleep(5000);
//            }
//            break;
//        }
//        return retu;
//    }
//
//    private String GetColumnName(String getMongodbName) throws IOException, InterruptedException {
//        HashMap<String, String> h = RuntimeEnv.TABLE_CR_COLUMNS.get(tableSe2DBRule.getRuleName());
//
//        if (h.get(getMongodbName) != null) {
//            return h.get(getMongodbName);
//        } else {
//            return null;
//        }
//    }
//
//    private boolean push_single_json_data(List<Map<String, Object>> jsonList) throws IOException {
//        boolean retu = true;
//        try {
//            long pp = 0l;
//            logger.info("jsonList  size is " + jsonList.size());
//            BoundStatement bs;
//            List<ResultSet> future = new ArrayList<>();
//
//            List<Map<String, Object>> errjson = new ArrayList<Map<String, Object>>();
//            for (Map<String,Object> str : jsonList) {
//                // 绑定数据并创建BoundStatement对象
//                Long ut = System.currentTimeMillis()/1000;
//                str.put("ut", ut);
//                String json = objectMapper.writeValueAsString(str);
//                bs = insert_statement.bind(json);
//                // 设置记录时间�?
//                if (this.timestampKey.length() > 0 && ((Map) str).containsKey(this.timestampKey)) {
//                    bs.setDefaultTimestamp((Long) ((Map) str).get(this.timestampKey) * 1000000); // second to microsecond
//                }//            batchStatement.add(bs);
//                // 同步插入
//                try {
//                    ResultSet result = sessionUI.execute(bs);
//                    String result_string = result.toString();
//                    if (result_string.contains("ERROR")) {
//                        logger.info(result_string);
//                        logger.info(json);
//                    }
//                    pp++;
//
//                } catch (OperationTimedOutException | QueryExecutionException | NoHostAvailableException e) {
//                    logger.error("Exception " + e.getMessage(),e);
//                    retu = false;
//                } catch (InvalidQueryException e) {
//                    logger.error("Exception " + e.getMessage(),e);
//                    // 此处有可能是客户端sql语句不正�?
//                    retu = false;
//                } catch (Exception E) {
//                    logger.error("Exception " + E);
//                    retu = false;
//                }
//
//            }
//            long endTime = System.nanoTime();
//            logger.info("0st load " + jsonList.size() + "|xxpect: " + pp + " records to table " + utableName + " successfully using " + (endTime - startTime) / 1000000L + " ms errorSIze: " + errjson.size() + "Boolean " + retu);
//            pp = 0l;
//            future.clear();
//
//        } catch (Exception E) {
//            logger.info(E);
//        }
//        return retu;
//    }
//
//    private Type resolveType(DataType rawType) {
//        switch (rawType.getName()) {
//            case ASCII:
//            case VARCHAR:
//            case TEXT:
//                return Type.STRING;
//            case TINYINT:
//                return Type.TINYINT;
//            case SMALLINT:
//                return Type.SMALLINT;
//            case INT:
//                return Type.INT;
//            case BIGINT:
//            case VARINT:
//                return Type.BIGINT;
//            case FLOAT:
//                return Type.FLOAT;
//            case DOUBLE:
//                return Type.DOUBLE;
//            case BLOB:
//                return Type.BLOB;
//            case BOOLEAN:
//                return Type.BOOLEAN;
//            case LIST:
//                return Type.LIST;
//            case MAP:
//                return Type.MAP;
//            case SET:
//                return Type.SET;
//            default:
//                throw new IllegalArgumentException("Unsupported type: " + rawType);
//
//        }
//    }
//
//    private enum Type {
//        STRING {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else {
//                    bs.setString(col, val.toString());
//                }
//            }
//        }, TINYINT {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof String) {
//                    bs.setByte(col, Long.valueOf(((String) val).toLowerCase()).byteValue());
//                } else if (val instanceof Number) {
//                    bs.setByte(col, ((Number) val).byteValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of " + col);
//                }
//            }
//        }, SMALLINT {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof String) {
//                    bs.setShort(col, Long.valueOf(((String) val).toLowerCase()).shortValue());
//                } else if (val instanceof Number) {
//                    bs.setShort(col, ((Number) val).shortValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of " + col);
//                }
//            }
//        }, INT {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof String) {
//                    bs.setInt(col, Long.valueOf(((String) val).toLowerCase()).intValue());
//                } else if (val instanceof Number) {
//                    bs.setInt(col, ((Number) val).intValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of " + col);
//                }
//            }
//        }, BIGINT {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof String) {
//                    bs.setLong(col, Long.valueOf(((String) val).toLowerCase()));
//                } else if (val instanceof Number) {
//                    bs.setLong(col, ((Number) val).longValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of " + col);
//                }
//            }
//        }, FLOAT {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof String) {
//                    bs.setFloat(col, Long.valueOf(((String) val).toLowerCase()).floatValue());
//                } else if (val instanceof Number) {
//                    bs.setFloat(col, ((Number) val).floatValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of " + col);
//                }
//            }
//        }, DOUBLE {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof String) {
//                    bs.setDouble(col, Long.valueOf(((String) val).toLowerCase()).doubleValue());
//                } else if (val instanceof Number) {
//                    bs.setDouble(col, ((Number) val).doubleValue());
//                } else {
//                    throw new IllegalArgumentException("wrong type of " + col);
//                }
//            }
//        }, BLOB {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof String) {
//                    try {
//                        byte[] bytes = Hex.decodeHex(((String) val).toCharArray());
//                        bs.setBytes(col, ByteBuffer.wrap(bytes));
//                    } catch (DecoderException e) {
//                        throw new IllegalArgumentException("wrong value of hex string, column: " + col);
//                    }
//                } else if (val instanceof byte[]) {
//                    bs.setBytes(col, ByteBuffer.wrap((byte[]) val));
//                } else if (val instanceof ByteBuffer) {
//                    bs.setBytes(col, (ByteBuffer) val);
//                } else {
//                    throw new IllegalArgumentException("wrong type of " + col);
//                }
//            }
//        }, BOOLEAN {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof String) {
//                    bs.setBool(col, Boolean.valueOf((String) val));
//                } else if (val instanceof Boolean) {
//                    bs.setBool(col, (Boolean) val);
//                } else {
//                    throw new IllegalArgumentException("wrong type of " + col);
//                }
//            }
//        }, LIST {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof List) {
//                    bs.setList(col, (List<?>) val);
//                } else if (val instanceof Collection) {
//                    List<?> l = new ArrayList<>((Collection<?>) val);
//                    bs.setList(col, l);
//                } else {
//                    List<Object> l = new ArrayList<>();
//                    l.add(val);
//                    bs.setList(col, l);
//                }
//            }
//        }, MAP {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof Map) {
//                    bs.setMap(col, (Map<?, ?>) val);
//                } else {
//                    throw new IllegalArgumentException("wrong type of " + col);
//                }
//            }
//        }, SET {
//            @Override
//            public void bind(String col, Object val, BoundStatement bs) {
//                if (val == null) {
//                    bs.setToNull(col);
//                } else if (val instanceof Set) {
//                    bs.setSet(col, (Set<?>) val);
//                } else if (val instanceof Collection) {
//                    Set<?> s = new HashSet<>((Collection<?>) val);
//                    bs.setSet(col, s);
//                } else {
//                    Set<Object> s = new HashSet<>();
//                    s.add(val);
//                    bs.setSet(col, s);
//                }
//            }
//        };
//
//        public void bind(String col, Object val, BoundStatement bs) {
//            throw new IllegalArgumentException("wrong type of " + col);
//        }
//    }
//
//    /**
//     * 提交更新数据到Cassandra
//     *
//     * @param lists 数据列表
//     */
//    public void close() {
//        try {
//            if (session != null || cluster != null) {
//                this.session.close();
//                this.cluster.close();
//            }
//        } catch (Exception E) {
//            logger.info(E);
//        } finally {
//            session = null;
//            cluster = null;
////            insert_statement = null;
//            update_statment = null;
//        }
//    }
//
//    public void uclose() {
//        try {
//            if (sessionUI != null || clusterUI != null) {
//                this.sessionUI.close();
//                this.clusterUI.close();
//            }
//        } catch (Exception E) {
//            logger.info(E);
//        } finally {
//            sessionUI = null;
//            clusterUI = null;
//            insert_statement = null;
//        }
//    }
//
//    /**
//     * 映射Cassandra类型到{@linkplain Type}枚举
//     *
//     * @param rawType Cassandra类型对象
//     * @return 返回Cassandra类型对应的Type枚举
//     */
//    public Session getSession() {
//        return this.session;
//    }
//    //去重程序  调用去重服务
//
//    private Object resolveType(Type tt, Object o) {
//        switch (tt) {
//            case STRING:
//                return o.toString();
//            case TINYINT:
//                return Type.TINYINT;
//            case SMALLINT:
//                return Type.SMALLINT;
//            case INT:
//                return Integer.valueOf(o.toString());
//            case BIGINT:
//                return Long.valueOf(o.toString());
//            case FLOAT:
//                return Type.FLOAT;
//            case DOUBLE:
//                return Type.DOUBLE;
//            case BLOB:
//                return Type.BLOB;
//            case BOOLEAN:
//                return Type.BOOLEAN;
//            case LIST:
//                return o;
//            case MAP:
//                return Type.MAP;
//            case SET:
//                return Type.SET;
//            default:
//                throw new IllegalArgumentException("Unsupported type: " + tt);
//        }
//    }
//
//    /**
//     * 设置更新条件
//     *
//     * @return
//     */
//    private String getUpdateStr(Set<String> upFields) {
//        StringBuilder sb = new StringBuilder();
//        // 遍历�?��更新的各个字段并写入where条件�?
//        for (String updateField : upFields) {
//            sb.append(updateField)
//                    .append("=:")
//                    .append(updateField)
//                    .append(",");
//        }
//        sb.deleteCharAt(sb.length() - 1);
//        return sb.toString();
//    }
//
//    /**
//     * 设置更新列的�?
//     *
//     * @param map 从消息队列中读取的数�?
//     * @param bs
//     */
//    private void setUpdateValue(Map map, BoundStatement bs,Set<String> feilds) {
//        Set<String> f = new HashSet();
//        f.addAll(feilds);
//        f.remove("key");
//        // 遍历�?��更新的各个列并分别设置更新�?
//        for (String updateField : f) {
//            Type type = columnMap.get(updateField);
//            Object val = map.get(updateField);
//            type.bind(updateField, val, bs);
//        }
//    }
//
//    private String getUpdatesql(String querykey, Set<String> feilds) {
//        Set<String> f = new HashSet();
//        f.addAll(feilds);
//        f.remove("key");
//        String[] querykeys = querykey.split(":");
//        int length = querykey.split(":").length;
//        String sql = "update " + tableName + " set "
//                + getUpdateStr(f) + " where ";
//        StringBuilder sb = new StringBuilder();
//        sb.append(sql);
//        int i = 0;
//        for (String s : querykeys) {
//            sb.append(s);
//            i++;
//            sb.append("=:" + "key" + " and ");
//        }
//        sb.append("seqn=:" + "seqn" + " and ");
//        int j = sb.lastIndexOf(" and");
//        String ss2 = sb.substring(0, j) + " if exists";
//        return ss2;
//    }
//
//    private void setKeys(String querykey, Map m, BoundStatement bs) {
//        int idx = 0;
//        String[] querykeys = querykey.split(":");
//        int i = 0;
//        for (String s : querykeys) {
//            i++;
//            columnMap.get(s).bind("key" + i, m.get(s), bs);
//        };
//    }
//
//}
